"""
Case Type   : reliability BM25 index
Case Name   : 存在索引场景下，插入事务（1个事务内1000条数据），未提交前取消插入事务，主备循环查询打分
Create At   : 2025/11/5
Owner       : chen
Description :
    1.创建表
    2.插入数据
    3.创建索引
    4.再次插入数据
    5.主机循环进行索引查询
    6.备机循环进行索引查询
    7.步骤4执行过程中，主机数据库取消创建索引进程；
    8.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.执行无报错
    6.执行无报错
    7.成功
    8.成功
History     :
"""

import os
import time
import unittest

from yat.test import Node
from yat.test import macro

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread
from testcase.utils.Logger import Logger

log = Logger()
primary_sh = CommonSH('PrimaryDbUser')

class reliabability_bm25_index_case0026(unittest.TestCase):
    def setUp(self):
        log.info(f'--------{os.path.basename(__file__)} start--------')
        self.constant = Constant()
        self.com = Common()
        self.pri_node = Node('PrimaryDbUser')
        self.sta_sh = CommonSH('PrimaryDbUser')
        self.table_name = 't_relia_bm25_index_026'
        self.index_name = 'idx_t_relia_bm25_index_026'
        self.pkg_csv = 'bm25_csv.tar.gz'
        self.csv1w_path = os.path.join(macro.DB_SCRIPT_PATH, 'copy_10000.csv')
        self.csv1k_path = os.path.join(macro.DB_SCRIPT_PATH, 'copy_1000.csv')

    def test_index(self):
        text = '--------step1: 创建表; expect: 成功--------'
        log.info(text)
        cr_tab = f'''drop table if exists {self.table_name};
            create table {self.table_name} 
            (id bigserial primary key,val_text text);
            alter table {self.table_name} set (parallel_workers=32);'''
        res = primary_sh.execut_db_sql(cr_tab)
        log.info(res)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, res)
        self.assertIn(self.constant.ALTER_TABLE_MSG, res)

        text = '--------step2: 插入数据; expect: 成功--------'
        log.info(text)
        check_cmd = f'if [ -f {self.csv1w_path} ]; '\
            f'then echo "True"; else echo "False"; fi'
        log.info(check_cmd)
        res = self.pri_node.sh(check_cmd).result()
        log.info(res)
        if res == 'False':
            self.com.wget_file(node=self.pri_node, 
                put_path=macro.DB_SCRIPT_PATH, 
                get_path=os.path.join(macro.FTP_PLUGINS_PATH,
                    self.pkg_csv))
            cmd = f"cd {macro.DB_SCRIPT_PATH} && tar -zxf {self.pkg_csv}"
            log.info(cmd)
            res = self.pri_node.sh(cmd).result()
            log.info(res)
        copy_sql = f"""copy {self.table_name} (val_text) from 
                '{self.csv1w_path}' with(DELIMITER '|');"""
        res = primary_sh.execut_db_sql(copy_sql)
        log.info(res)
        self.assertIn(self.constant.copy_success_msg, res)

        text = '--------step3: 创建索引; expect:开始执行--------'
        log.info(text)
        cr_ix = f'create index {self.index_name} on ' \
            f'{self.table_name} using bm25(val_text);'
        res = primary_sh.execut_db_sql(cr_ix)
        log.info(res)
        self.assertIn(self.constant.CREATE_INDEX_SUCCESS, res)

        text = '--------step4: 再次插入数据; expect:开始执行--------'
        log.info(text)
        copy_sql = f"""copy {self.table_name} (val_text) from 
                '{self.csv1k_path}' with(DELIMITER '|');"""
        threads = []
        step4_thread = ComThread(primary_sh.execut_db_sql,
                                        args=(copy_sql,))
        threads.append(step4_thread)

        text = '--------step5: 主机循环进行索引查询; expect:无报错--------'
        log.info(text)
        find_sql = f'set enable_indexscan=on;' \
            f'set enable_seqscan=off;' \
            f"select id,val_text <&> '《后汉书》与钢铁是怎样练成的文章内容比较' from " \
            f"{self.table_name} order by val_text <&> " \
            f"'《后汉书》与钢铁是怎样练成的文章内容比较' desc limit 5;"
        step5_thread = ComThread(self.cycle_exec,
                                    args=('主机', find_sql, 100, primary_sh))
        threads.append(step5_thread)

        text = '--------step6: 备机循环进行索引查询; expect:无报错--------'
        log.info(text)
        step6_thread = ComThread(self.cycle_exec,
                                    args=('备机', find_sql, 100, self.sta_sh))
        threads.append(step6_thread)

        text = '--------step7: 主机数据库取消创建索引进程; expect:成功--------'
        log.info(text)
        for t in threads:
            t.setDaemon(True)
            t.start()
        time.sleep(10)
        cmd = f"ps ux|grep 'copy {self.table_name}'|grep -v grep|awk " \
            r"'{{print $2}}'|xargs kill -2"
        log.info(cmd)
        res = self.pri_node.sh(cmd).result()
        log.info(res)
        for t in threads:
            t.join()
        self.assertEqual(step5_thread.get_result(), step6_thread.get_result())

    def cycle_exec(self, hint, sql, times, node_sh):
        result = ''
        for i in range(times):
            log.info(f"{hint}循环第{i + 1}次查询")
            result = node_sh.execut_db_sql(sql)
            log.info(result)
            self.assertTrue('Error' not in result or 
                'error' not in result or 'ERROR' not in result)
        return result

    def tearDown(self):
        text = '--------step8: 清理环境; expect:成功--------'
        log.info(text)
        clear_sql = f'drop table if exists {self.table_name};'
        res = primary_sh.execut_db_sql(clear_sql)
        log.info(res)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, res)